package cn.krguang.gasalermjiangshan;

import android.util.Log;

import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;


public class MyMqttClient {
    private static final String TAG  = "test123";
    private static MyMqttClient myMqttClient;

    private static String MqttUserString = "krg";
    private static String MqttPwdString = "Mqtt17612168015";
    private static String MqttIPString = "www.krguang.cn";
    private static int MqttPort = 1883;

    private static final String TOPIC = "device/gas/jiangshanzhongyiyuan/#";

    private static MqttClient mqttClient;
    private static MqttConnectOptions mqttConnectOptions;
    private boolean ConnectFlage = true;

    private OnServerConnectedCallback ConnectedCallback;//连接到服务器
    private OnServerDisConnectedCallback DisConnectedCallback;//与服务器断开连接
    private OnServerReadStringCallback ReadStringCallback;//接收信息回调(字符串)

    // 创建一个单任务线程池
    private ExecutorService pool;

    private MyMqttClient() {//构造函数私有化
        super();
        pool = Executors.newSingleThreadExecutor();
    }
    //提供一个全局的静态方法
    public static MyMqttClient sharedCenter() {
        if (myMqttClient == null) {
            synchronized (MyMqttClient.class) {
                if (myMqttClient == null) {
                    myMqttClient = new MyMqttClient();
                    mqttClient = null;
                }
            }
        }
        return myMqttClient;
    }

    private void InitMqttOptions(){
        mqttConnectOptions = new MqttConnectOptions();//MQTT的连接设置
        //设置是否清空session,这里如果设置为false表示服务器会保留客户端的连接记录，这里设置为true表示每次连接到服务器都以新的身份连接
        mqttConnectOptions.setCleanSession(true);
        mqttConnectOptions.setUserName(MqttUserString);//设置连接的用户名
        mqttConnectOptions.setPassword(MqttPwdString.toCharArray());//设置连接的密码
        mqttConnectOptions.setConnectionTimeout(3);// 设置连接超时时间 单位为秒
        // 设置会话心跳时间 单位为秒
        mqttConnectOptions.setKeepAliveInterval(60);
    }

    /*初始化Mqtt连接*/
    private void InitMqttConnect() {
        try {
            long time= System.currentTimeMillis();
            String Str = time+"";
            Str = Str.substring(Str.length()-4,Str.length());
            mqttClient = new MqttClient("tcp://"+MqttIPString+":"+MqttPort,Str,new MemoryPersistence());
            mqttClient.setCallback(new MqttCallback() {
                @Override
                public void messageArrived(String arg0, MqttMessage arg1){
                 //   getNetStatus(arg0, arg1);
                //    getAndProcessData(arg0, arg1);

                    if (ReadStringCallback!=null){
                        ReadStringCallback.callback(arg0,arg1);
                    }
                }
                @Override
                public void deliveryComplete(IMqttDeliveryToken arg0) { }
                @Override
                public void connectionLost(Throwable arg0) {
                    if (DisConnectedCallback!=null)
                        DisConnectedCallback.callback(arg0);
                    try {
                        mqttClient.disconnect();
                    }
                    catch (Exception e)
                    {
                        e.printStackTrace();
                    }
                    try {
                        mqttClient.close();
                    } catch (Exception e)
                    {
                        e.printStackTrace();
                    }
                    mqttClient = null;
                    ConnectFlage = true;
                    setConnect();
                }
            });
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void setConnect(){
        if (mqttClient == null || !mqttClient.isConnected()) {
            InitMqttConnect();
            InitMqttOptions();
        }else {
             Log.d(TAG, "setConnect: mqttClient Is Connected" );
        }
        pool.execute(new Thread(new Runnable() {
            @Override
            public void run() {
                while(ConnectFlage) {
                    try {
                        if (mqttClient == null || !mqttClient.isConnected()) {
                               Log.d(TAG, "run: Connect ......");
                            mqttClient.connect(mqttConnectOptions);
                               Log.d(TAG, "run: Connect Success??");
                            if (mqttClient.isConnected()) {
                                if (ConnectedCallback != null){
                                    ConnectedCallback.callback();
                                }
                                ConnectFlage = false;
                                Log.d(TAG, "run: Connect Success");
                                setSubscribe(TOPIC, 0);
                            }
                        }
                    } catch (Exception e) {
                        e.printStackTrace();
                        if (DisConnectedCallback!=null)
                            DisConnectedCallback.callback(e);
                        try{
                            Thread.sleep(3000);
                        }catch (Exception e1){
                            e1.printStackTrace();
                        }
                    }
                }
            }
        }));
    }

    //订阅主题
    public void setSubscribe(final String Topic, final int qos) {
        pool.execute(new Thread(new Runnable() {
            @Override
            public void run() {
                try{
                    if (mqttClient!=null){
                        mqttClient.subscribe(Topic,qos);
                    }
                }
                catch (MqttException e) {
                    e.printStackTrace();
                }
            }
        }));
    }
    //取消订阅主题
    public void setUnSubscribe(final String Topic) {
        pool.execute(new Thread(new Runnable() {
            @Override
            public void run() {
                try{
                    if (mqttClient!=null){
                        mqttClient.unsubscribe(Topic);
                    }
                }
                catch (MqttException e){
                    e.printStackTrace();
                }
            }
        }));
    }

    public void setDisConnect() {
        pool.execute(new Thread(new Runnable() {
            @Override
            public void run() {
                   Log.d(TAG, "run: DisConnect");
                if (mqttClient!=null){
                    try { mqttClient.disconnect();} catch (Exception e) {e.printStackTrace();}
                    try {mqttClient.close();} catch (Exception e) {e.printStackTrace();}
                    myMqttClient = null;
                    Log.d(TAG, "setDisConnect: OK");
                }
            }
        }));
    }

    public void setRstConnect()
    {
        pool.execute(new Thread(new Runnable() {
            @Override
            public void run() {
                try { mqttClient.disconnect();} catch (Exception e) {
                    e.printStackTrace();
                }
                try {mqttClient.close();} catch (Exception e) {
                    e.printStackTrace();
                }
                mqttClient = null;
                ConnectFlage = true;
                setConnect();
            }
        }));
    }

    //发送数据
    public void setSendData(final String Topic, final byte[] bytes) {
        pool.execute(new Thread(new Runnable() {
            @Override
            public void run() {
                MqttMessage msgMessage = new MqttMessage(bytes);
                try {
                    if (mqttClient!=null){
                        mqttClient.publish(Topic,msgMessage);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }));
    }

    //发送数据
    public void setSendData(final String Topic, final String bytes) {
        pool.execute(new Thread(new Runnable() {
            @Override
            public void run() {
                try {
                    if (mqttClient!=null) {
                        MqttMessage msgMessage = new MqttMessage(bytes.getBytes());
                        mqttClient.publish(Topic,msgMessage);
                    }
                }
                catch (Exception e)
                {
                    e.printStackTrace();
                }
            }
        }));
    }

    public interface OnServerConnectedCallback {
        void callback();
    }
    public interface OnServerDisConnectedCallback {
        void callback(Throwable e);
    }
    public interface OnServerReadStringCallback {
        void callback(String Topic, MqttMessage Msg);
    }

    public void setOnServerConnectedCallback(OnServerConnectedCallback ConnectedCallback) {
        this.ConnectedCallback = ConnectedCallback;
    }

    public void setOnServerDisConnectedCallback(OnServerDisConnectedCallback DisConnectedCallback) {
        this.DisConnectedCallback = DisConnectedCallback;
    }

    public void setOnServerReadStringCallback(OnServerReadStringCallback ReadStringCallback) {
        this.ReadStringCallback = ReadStringCallback;
    }
}